package com.chukun.flink.table.api.sql;

import com.chukun.flink.table.bean.ClickBean;
import com.chukun.flink.table.bean.Person;
import com.chukun.flink.table.source.PrepareData;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import java.sql.Timestamp;
import java.util.Collections;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author chukun
 * @version 1.0.0
 * @description join基本操作
 * @createTime 2022年06月03日 10:27:00
 */
public class JoinTemplate {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStream<ClickBean> clickStream = env.fromCollection(PrepareData.getClicksData());

        clickStream = clickStream.map(new MapFunction<ClickBean, ClickBean>() {
            @Override
            public ClickBean map(ClickBean clickBean) throws Exception {
                TimeUnit.SECONDS.sleep(2);
                return clickBean;
            }
        });

        tableEnv.createTemporaryView("Clicks", clickStream,  Schema.newBuilder()
                .column("user", DataTypes.STRING())
                .column("url", DataTypes.STRING())
                .column("time", DataTypes.TIMESTAMP().bridgedTo(Timestamp.class))
                .build());

        List<Person> personData = PrepareData.getPersonData();

        Collections.shuffle(personData);

        DataStream<Person> personStream = env.fromCollection(personData);

        personStream = personStream.map(new MapFunction<Person, Person>() {
            @Override
            public Person map(Person person) throws Exception {
                TimeUnit.SECONDS.sleep(4);
                return person;
            }
        });

        tableEnv.createTemporaryView("Person", personStream,  Schema.newBuilder()
                .column("name", DataTypes.STRING())
                .column("age", DataTypes.BIGINT())
                .column("city", DataTypes.STRING())
                .build());

        // inner join
        String innerJoinSql = "select * from Clicks inner join Person on name=user";

        Table innerJoinResult = tableEnv.sqlQuery(innerJoinSql);

        DataStream<Row> innerJoinDataStream = tableEnv.toChangelogStream(innerJoinResult);

        innerJoinDataStream.print("inner-join");

        // left join
        String leftJoinSql = "select * from Clicks left join Person on name=user";

        Table leftJoinresult = tableEnv.sqlQuery(leftJoinSql);

        DataStream<Row> leftJoinDataStream = tableEnv.toChangelogStream(leftJoinresult);

        leftJoinDataStream.print("left-join");

        // right join
        String rightJoinSql = "select * from Clicks right join Person on name=user";

        Table rightJoinresult = tableEnv.sqlQuery(rightJoinSql);

        DataStream<Row> rightJoinDataStream = tableEnv.toChangelogStream(rightJoinresult);

        rightJoinDataStream.print("right-join");

        // full join
        String fullJoinSql = "select * from Clicks full join Person on name=user";

        Table fullJoinresult = tableEnv.sqlQuery(fullJoinSql);

        DataStream<Row> fullJoinDataStream = tableEnv.toChangelogStream(fullJoinresult);

        fullJoinDataStream.print("full-join");

        env.execute("JoinTemplate");

    }
}
